package com.atguigu.flink.chapter07;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/23 16:15
 */
public class MyPeriodicWatermarkGenerator<T> implements WatermarkGenerator<T> {

    private long maxTimestamp = -1L;

    /**
     * 每来一条数据数据，调用一次该方法
     *
     * @param event          一个事件（一条数据）
     * @param eventTimestamp 数据的事件时间
     * @param output
     */
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        System.out.println("on Event...");
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
    }

    /**
     * 每隔一个周期，调用一次该方法
     *
     * @param output
     */
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        System.out.println("on periodic...");
        // 发射watermark
        output.emitWatermark(new Watermark(maxTimestamp));
    }
}
